package cn.ctyun.datax.plugin.reader.httpreader.util;

import cn.ctyun.datax.plugin.reader.httpreader.constant.*;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.jayway.jsonpath.JsonPath;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.client.RestClientException;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.regex.Pattern;

/**
 * HTTP工具类
 *
 * @author cuijianpeng
 * @date 2020/5/20 22:32
 */
public class HttpUtil {

    private static final Logger LOG = LoggerFactory.getLogger(HttpUtil.class);

    public static String urlRequest(Configuration config) {
        String httpMethod = config.getString(KeyConstant.HTTP_METHOD);
        Long delayTime = config.getLong(KeyConstant.DELAY_TIME);
        int repeat = 0;
        while (repeat < HttpConstant.TOTAL_REPEAT) {
            sleep(delayTime);
            String result;
            try {
                result = HttpMethodEnum.valueOf(httpMethod.toUpperCase()).urlRequest(config);
            } catch (RestClientException e) {
                repeat++;
                if (repeat < HttpConstant.TOTAL_REPEAT) {
                    sleep(delayTime * 5 * repeat + 1);
                    LOG.debug(String.format("url访问失败，启动重试[ %s / %s ]，失败原因：%s",
                            repeat + 1, HttpConstant.TOTAL_REPEAT, e.getMessage()));
                }
                continue;
            }
            if (StringUtils.isBlank(result)) {
                throw DataXException.asDataXException(HttpReaderErrorCode.RESPONSE_NULL, "url响应数据为空");
            }
            String codeJsonPath = config.getString(KeyConstant.CODE_JSON_PATH);
            Object code = JsonPath.read(result, codeJsonPath);
            Object trueCode = config.get(KeyConstant.CODE);
            LOG.debug(String.format("判断返回值是否成功[预期值：%s，类型：%s],[实际值：%s，类型：%s]",
                    trueCode, trueCode.getClass().toString(), code, code.getClass().toString()));
            if (!trueCode.equals(code)) {
                throw DataXException.asDataXException(HttpReaderErrorCode.RESPONSE_NULL,
                        String.format("url响应数据错误[%s]", result));
            }
            return result;
        }
        throw DataXException.asDataXException(HttpReaderErrorCode.REQUEST_ERROR,
                String.format("url请求失败[%s]", config.getString(KeyConstant.URL)));
    }

    /**
     * 根据url和header单线程分配Job
     *
     * @param config 配置信息
     * @return Jobs
     */
    public static List<Configuration> splitJob(Configuration config) {
        List<Configuration> splitConfigs = new ArrayList<>();
        JSONArray urlArray = JSON.parseArray(config.getString(KeyConstant.URL));
        final String headerStr = config.getString(KeyConstant.HEADER);
        JSONArray headerArray;
        if (StringUtils.isNotBlank(headerStr)) {
            headerArray = JSON.parseArray(headerStr);
        } else {
            headerArray = new JSONArray();
        }
        if (headerArray.isEmpty()) {
            headerArray.add(new JSONObject());
        }
        for (Object url : urlArray) {
            for (Object header : headerArray) {
                Configuration tempSlice = config.clone();
                tempSlice.set(KeyConstant.URL, url);
                tempSlice.set(KeyConstant.HEADER, ((JSONObject)header).toJSONString());
                splitConfigs.add(tempSlice);
            }
        }
        return splitConfigs;
    }

    public static List<Object> parseResponse(Configuration config, String response) {
        String dataJsonPath = config.getString(KeyConstant.DATA_JSON_PATH);
        final Object read = JsonPath.read(response, dataJsonPath);
        List<Object> data = new ArrayList<>();
        if (read instanceof List) {
            data.addAll(JSON.parseArray(read.toString()));
        } else {
            data.add(read);
        }
        if (data.isEmpty()) {
            throw DataXException.asDataXException(HttpReaderErrorCode.PARSE_ERROR,
                    String.format("JSON解析失败，解析结果为空，json:[%s]", response));
        }
        return data;
    }

    public static void transportRecord(Configuration config, List<Object> data,
                                       RecordSender recordSender, TaskPluginCollector taskPluginCollector) {
        JSONArray columnMeta = JSON.parseArray(config.getString(KeyConstant.HTTP_COLUMN));
        for (Object datum : data) {
            Record record = recordSender.createRecord();
            try {
                final JSONObject item = JSON.parseObject(String.valueOf(datum));
                for (Object column : columnMeta) {
                    Object columnName = ((JSONObject) column).getString(KeyConstant.COLUMN_NAME);
                    Object columnType = ((JSONObject) column).getString(KeyConstant.COLUMN_TYPE);
                    record.addColumn(ColumnTypeEnum.valueOf(String.valueOf(columnType).toUpperCase()).getColumnValue(item, columnName));
                }
            } catch (Exception var11) {
                taskPluginCollector.collectDirtyRecord(record, var11);
                if (var11 instanceof DataXException) {
                    throw (DataXException) var11;
                }
            }
            recordSender.sendToWriter(record);
        }
    }

    /**
     * 简单判断是否为JSONArray
     *
     * @param content 字符串
     * @return 是否为JSONArray
     */
    private static boolean isJsonArray(String content) {
        return Objects.nonNull(content) && Pattern.matches(HttpConstant.SQUARE_REGEX, content);
    }

    /**
     * url请求休眠
     *
     * @param delayTime 休眠时间
     */
    private static void sleep(Long delayTime) {
        try {
            Thread.sleep(delayTime);
        } catch (InterruptedException ignored) {
        }
    }

}
